{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "previous-irish",
   "metadata": {
    "papermill": {
     "duration": 0.005346,
     "end_time": "2021-03-17T17:21:55.111968",
     "exception": false,
     "start_time": "2021-03-17T17:21:55.106622",
     "status": "completed"
    },
    "tags": []
   },
   "source": [
    "# Converts a parquet file to CSV file with header using ApacheSpark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "palestinian-toilet",
   "metadata": {
    "papermill": {
     "duration": 2.451598,
     "end_time": "2021-03-17T17:21:57.568763",
     "exception": false,
     "start_time": "2021-03-17T17:21:55.117165",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%bash\n",
    "export version=`python --version |awk '{print $2}' |awk -F\".\" '{print $1$2}'`\n",
    "\n",
    "echo $version\n",
    "\n",
    "if [ $version == '36' ] || [ $version == '37' ]; then\n",
    "    echo 'Starting installation...'\n",
    "    pip3 install pyspark==2.4.8 wget==3.2 pyspark2pmml==0.5.1 > install.log 2> install.log\n",
    "    if [ $? == 0 ]; then\n",
    "        echo 'Please <<RESTART YOUR KERNEL>> (Kernel->Restart Kernel and Clear All Outputs)'\n",
    "    else\n",
    "        echo 'Installation failed, please check log:'\n",
    "        cat install.log\n",
    "    fi\n",
    "elif [ $version == '38' ] || [ $version == '39' ]; then\n",
    "    pip3 install pyspark==3.1.2 wget==3.2 pyspark2pmml==0.5.1 > install.log 2> install.log\n",
    "    if [ $? == 0 ]; then\n",
    "        echo 'Please <<RESTART YOUR KERNEL>> (Kernel->Restart Kernel and Clear All Outputs)'\n",
    "    else\n",
    "        echo 'Installation failed, please check log:'\n",
    "        cat install.log\n",
    "    fi\n",
    "else\n",
    "    echo 'Currently only python 3.6, 3.7 , 3.8 and 3.9 are supported, in case you need a different version please open an issue at https://github.com/IBM/claimed/issues'\n",
    "    exit -1\n",
    "fi"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "soviet-tobacco",
   "metadata": {
    "papermill": {
     "duration": 0.009853,
     "end_time": "2021-03-17T17:21:57.584134",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.574281",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# @param data_dir temporal data storage for local execution\n",
    "# @param data_csv csv path and file name (default: data.csv)\n",
    "# @param data_parquet path and parquet file name (default: data.parquet)\n",
    "# @param master url of master (default: local mode)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "nearby-transmission",
   "metadata": {
    "papermill": {
     "duration": 0.132622,
     "end_time": "2021-03-17T17:21:57.721932",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.589310",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "import os\n",
    "import shutil\n",
    "import glob"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "productive-congress",
   "metadata": {
    "papermill": {
     "duration": 0.010157,
     "end_time": "2021-03-17T17:21:57.737545",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.727388",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "data_csv = os.environ.get('data_csv', 'data.csv')\n",
    "data_parquet = os.environ.get('data_parquet', 'data.parquet')\n",
    "master = os.environ.get('master', \"local[*]\")\n",
    "data_dir = os.environ.get('data_dir', '../../data/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "associate-sixth",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_parquet = 'trends.parquet'\n",
    "data_csv = 'trends.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ordinary-projection",
   "metadata": {
    "papermill": {
     "duration": 0.009268,
     "end_time": "2021-03-17T17:21:57.751975",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.742707",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "skip = False\n",
    "if os.path.exists(data_dir + data_csv):\n",
    "    skip = True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "indie-classics",
   "metadata": {
    "papermill": {
     "duration": 0.028188,
     "end_time": "2021-03-17T17:21:57.786591",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.758403",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "if not skip:\n",
    "    sc = SparkContext.getOrCreate(SparkConf().setMaster(master))\n",
    "    spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "finnish-shaft",
   "metadata": {
    "papermill": {
     "duration": 0.010092,
     "end_time": "2021-03-17T17:21:57.802836",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.792744",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "if not skip:\n",
    "    df = spark.read.parquet(data_dir + data_parquet)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "competitive-whole",
   "metadata": {
    "papermill": {
     "duration": 0.008981,
     "end_time": "2021-03-17T17:21:57.816796",
     "exception": false,
     "start_time": "2021-03-17T17:21:57.807815",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "if not skip:\n",
    "    if os.path.exists(data_dir + data_csv):\n",
    "        shutil.rmtree(data_dir + data_csv)\n",
    "    df.coalesce(1).write.option(\"header\", \"true\").csv(data_dir + data_csv)\n",
    "    file = glob.glob(data_dir + data_csv + '/part-*')\n",
    "    shutil.move(file[0], data_dir + data_csv + '.tmp')\n",
    "    shutil.rmtree(data_dir + data_csv)\n",
    "    shutil.move(data_dir + data_csv + '.tmp', data_dir + data_csv)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.6"
  },
  "papermill": {
   "default_parameters": {},
   "duration": 3.911738,
   "end_time": "2021-03-17T17:21:58.127228",
   "environment_variables": {},
   "exception": null,
   "input_path": "/home/jovyan/work/examples/pipelines/pairs/component-library/transform/spark-csv-to-parquet.ipynb",
   "output_path": "/home/jovyan/work/examples/pipelines/pairs/component-library/transform/spark-csv-to-parquet.ipynb",
   "parameters": {},
   "start_time": "2021-03-17T17:21:54.215490",
   "version": "2.3.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
